package com.xiaour.spring.boot.rocketmq.consumer;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Set;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

@Component
public class notifyConsumer {
	 /**
     * 消费者
     */
    @Value("${apache.rocketmq.consumer.pushConsumer}")
    private String pushConsumer;

    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    
    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.consumer.notifyConsumer}")
    private String consumerGroup;
    
    private DefaultMQPullConsumer consumer;
    
    @PostConstruct
    public void defaultMQProducer() {
    	
    	
        consumer = new DefaultMQPullConsumer(consumerGroup);
        
        consumer.setNamesrvAddr(namesrvAddr);
        

        try {
        	consumer.start();
            System.out.println("-------->:notifyConsumer启动了");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    public String viewMessage(String OffsetMsgId) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
    	 MessageExt msg = consumer.viewMessage(OffsetMsgId);
        return "{\"OffsetMsgId\":\""+OffsetMsgId+"\", \"msg\":\""+new String(msg.getBody())+"\"}";
    }
    
    
    public String pull(String topic, String keys, String offset) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        
    	
    	try {
        	
        	
    		JSONObject resultJson = new JSONObject();
    		
    		
    		Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);
    		if(mqs.size() != 1)
    		{
    			return "FRRE";
    		}
    		for (MessageQueue mq : mqs) {
                PullResult pullResult =
                        consumer.pullBlockIfNotFound(mq, null, Long.parseLong(offset), 32);
                    System.out.printf("%s%n", pullResult);
                    
                    resultJson.put("nextBeginOffset", pullResult.getNextBeginOffset());
                    resultJson.put("maxOffset", pullResult.getMaxOffset());
                    switch (pullResult.getPullStatus()) {
                    case FOUND:
                        List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                        resultJson.put("num", messageExtList.size());
                        int i=1;
                        for (MessageExt m : messageExtList) {
                            System.out.println(new String(m.getBody()));
                            resultJson.put(String.valueOf(i), new String(m.getBody()));
                            i++;
                        }
                        break;
                    case NO_MATCHED_MSG:
                    	resultJson.put("result", "NO_MATCHED_MSG");
                        break;
                    case NO_NEW_MSG:
                    	resultJson.put("result", "NO_NEW_MSG");
                        break;
                    case OFFSET_ILLEGAL:
                    	resultJson.put("result", "OFFSET_ILLEGAL");
                        break;
                    default:
                        break;
                }
    		}

            System.out.println(resultJson);
            

            resultJson.put("userid", keys);
            resultJson.put("topic", "notifyinfor");
            
            return resultJson.toString();
        } catch (Exception e) {
            e.printStackTrace();
        }
		return "ERROR";
        
      
        
        
   }

}

